package org.dialMsg.Producer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.dialMsg.Pojo.HnCrmContChildState;
import org.dialMsg.Pojo.TCrmContChildState;
import org.dialMsg.Pojo.TDialResult;
import org.dialMsg.Server.ServerImp.FactMessageSendApiServerImp;
import org.dialMsg.Server.ServerImp.JpaEntityServiceImp;
import org.dialMsg.Server.ServerImp.TDialResultImp;
import org.dialMsg.Server.ServerInt.RobotExecLogService;
import org.dialMsg.Topic.RocketmqConstant;
import org.dialMsg.Server.ServerImp.PostExampleHutool;
import org.dialMsg.Topic.SendTopicConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.text.SimpleDateFormat;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author DPF
 */
@Service
public class SendTopic {
    private static final Logger logger = LoggerFactory.getLogger(SendTopic.class);
    private final RedisTemplate<String, String> redisTemplate;
    private final RocketMQTemplate rocketMQTemplate;
    @Autowired
    private TDialResultImp tDialResultImp;
    @Autowired
    private JpaEntityServiceImp jpaEntityService;
    @Autowired
    private PostExampleHutool client;
    @Autowired
    private FactMessageSendApiServerImp factMessageSendApiServerImp;
    @Autowired
    private RobotExecLogService robotExecLogService;

    public SendTopic(RocketMQTemplate rocketMQTemplate, RedisTemplate<String, String> redisTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
        this.redisTemplate = redisTemplate;
    }

    public String sendMessage() throws InterruptedException {
        String lastMaxIdStr = redisTemplate.opsForValue().get("lastMaxId");
        Long lastMaxId = lastMaxIdStr != null ? Long.parseLong(lastMaxIdStr) : 828704L;
        String phone = "111222333444";

        List<TDialResult> tDialResults = tDialResultImp.queryIdByPhoneAfter(lastMaxId, phone);
        if (tDialResults.isEmpty()) {
            return "没有更多数据";
        }
        try {
            factMessageSendApiServerImp.sendFactMessage(tDialResults);
        } catch (Exception e) {
            logger.error("调用 factMessageSendApiServerImp.sendFactMessage 失败，忽略异常继续执行，error={}", e.getMessage(), e);
        }
        List<String> keysList = new ArrayList<>();
        Long currentMaxId = lastMaxId;

        String value = RocketmqConstant.MyTopic()
                .get(RocketmqConstant.TopicEnum.ROBOT_DIAL_RESULT_MSG_TOPIC);

        for (TDialResult tDialResult : tDialResults) {
            // 构造消息体
            JSONObject dialData = new JSONObject();
            dialData.put("dialId", tDialResult.getId());
            dialData.put("callType", tDialResult.getCallType());
            dialData.put("batchId", tDialResult.getDataSourceBatchId());
            dialData.put("phone", tDialResult.getPhone());
            dialData.put("sceneId", tDialResult.getBusinessId());
            dialData.put("caller", tDialResult.getOutboundNumber());
            dialData.put("roundCount", tDialResult.getRoundCount());
            dialData.put("robotId", tDialResult.getRobotId());
            dialData.put("transferResult", tDialResult.getBridgeResult());
            dialData.put("durationOfSeatConversing", tDialResult.getDurationOfSeatConversing());
            dialData.put("taskName", tDialResult.getShowName());
            dialData.put("batchName",  tDialResult.getShowName());
            dialData.put("dataSourceType", tDialResult.getCallType());
            dialData.put("bankManager", tDialResult.getBankManager());
            dialData.put("bankOrgName", tDialResult.getBusinessName());
            dialData.put("batchDetailId", tDialResult.getDataSourceBatchDetailId());
            dialData.put("bankOrg", tDialResult.getBankOrg());
            dialData.put("followUserId", tDialResult.getFollowUser());
            JSONArray dataArray = new JSONArray();
            dataArray.add(dialData);
            JSONObject mainObject = new JSONObject();
            mainObject.put("accountId",tDialResult.getAccountId());
            mainObject.put("data", dataArray);
            String dialTaskMQMsg = mainObject.toJSONString();
          //生成话单中的语音文字
            try {
                String content = "[{\"callContent\":\"喂（空1.2秒）您好，这里是珠海建行，您在我行有笔定期存款已经到期还未办理转存，我安排网点客户经理给您介绍下最新转存产品信息好吗？!(R130)\",\"callTime\":1739614767097,\"createTime\":\"2025-02-15 18:19:27\",\"flowNodeSeq\":\"R130\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"nodeId\":\"df831ec0-48dc-11ec-8ea2-2766b20a6b2d\",\"playbackUid\":\"bc68f93446cbd7d428ab52a4edec71bc\",\"robot\":true,\"skipStat\":false,\"tips\":\"开场白\",\"voiceCostTime\":0},{\"analyseType\":6,\"asrResult\":\"你说什么？我没听清。\",\"callContent\":\"你说什么？我没听清。(Q61)\",\"callContentCutPos\":\"你/r,说/v,什么/r,？/w,我/r,没/v,听/v,清/LOC,。/w\",\"callTime\":1739614785964,\"createTime\":\"2025-02-15 18:19:45\",\"flowNodeSeq\":\"Q61\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"qaEntity\":{\"highIntentionQa\":0,\"intentionTag\":\"What\",\"intentionType\":3,\"qaName\":\"227.tmpc_什么事\",\"qaUid\":61,\"qawords\":[{\"scroe\":0.5,\"word\":\"227.tmpc_什么事\"}],\"tagName\":\"227.tmpc_什么事\",\"targetWord\":\"你说#什么#\",\"type\":1},\"robot\":false,\"skipStat\":false,\"tips\":\"中性QA命中\",\"voiceCostTime\":0},{\"callContent\":\"是这样，来电是提醒您在我行有笔定期存款已经到期还未转存，建议您尽快办理转存，以免影响到您的收益。稍后我让客户经理给您具体介绍下转存产品信息及流程好吗？(R104)\",\"callTime\":1739614785964,\"createTime\":\"2025-02-15 18:19:45\",\"flowNodeSeq\":\"R104\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"playbackUid\":\"4997788b06bf0641fdd1d3a729e65e06\",\"robot\":true,\"skipStat\":false,\"voiceCostTime\":0},{\"analyseType\":7,\"asrResult\":\"你再说一遍，我没听清。\",\"callContent\":\"你再说一遍，我没听清。(Y4)\",\"callContentCutPos\":\"你/r,再说/d,一遍/m,，/w,我/r,没/v,听/v,清/LOC,。/w\",\"callTime\":1739614803510,\"createTime\":\"2025-02-15 18:20:03\",\"flowNodeSeq\":\"Y4\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"robot\":false,\"skipStat\":false,\"tips\":\"语态识别命中\",\"toneEntity\":{\"attribute\":6,\"attributeCategoryName\":\"sys.重复\",\"attributeCategoryUid\":4,\"targetWord\":\"Model\",\"tonewords\":[{\"scroe\":0.9999995231628418,\"value\":6,\"word\":\"sys.重复\"}]},\"voiceCostTime\":0},{\"callContent\":\"是这样，来电是提醒您在我行有笔定期存款已经到期还未转存，建议您尽快办理转存，以免影响到您的收益。稍后我让客户经理给您具体介绍下转存产品信息及流程好吗？(R104)\",\"callTime\":1739614803510,\"createTime\":\"2025-02-15 18:20:03\",\"flowNodeSeq\":\"R104\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"playbackUid\":\"4997788b06bf0641fdd1d3a729e65e06\",\"robot\":true,\"skipStat\":false,\"voiceCostTime\":0},{\"analyseType\":1,\"callContent\":\"被叫挂断(UE)\",\"callTime\":1739614814465,\"createTime\":\"2025-02-15 18:20:14\",\"flowNodeSeq\":\"UE\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"robot\":false,\"skipStat\":false,\"tips\":\"被叫挂断\",\"voiceCostTime\":0,\"voiceText\":\"被叫挂断\"}]";
                robotExecLogService.addLog(tDialResult.getId(), content, tDialResult.getTimeCreate());
            } catch (Exception e) {
                logger.error("生成话单中的语音文字时出错，忽略异常继续执行", e);
            }
            // 生成KEYS并发送MQ消息（先发送MQ）
            String KEYS = String.format("%s%s",
                    new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()),
                    RandomStringUtils.randomAlphabetic(18).toUpperCase()
            );
            keysList.add(KEYS);
            try {
                rocketMQTemplate.send(value,
                        MessageBuilder.withPayload(dialTaskMQMsg)
                                .setHeader("KEYS", KEYS)
                                .build()
                );
            } catch (Exception e) {
                logger.error("发送RocketMQ消息失败，dialId={}, error={}", tDialResult.getId(), e.getMessage(), e);
            }
            // 执行post请求（调整到MQ发送之后）
            int[] statusArray = {11, 20, 21};
            int randomIndex = RandomUtils.nextInt(0, statusArray.length);
            int randomStatus = statusArray[randomIndex];
            try {
                client.postExample(
                        tDialResult.getBusinessId(),
                        randomStatus,
                        phone,
                        tDialResult.getCallType(),
                        new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(tDialResult.getTimeCreate()),
                        tDialResult.getBankManager(),
                        tDialResult.getId()
                );
            } catch (Exception e) {
                logger.error("postExample请求失败，dialId={}, error={}", tDialResult.getId(), e.getMessage(), e);
            }

            // 更新最大ID
            if (tDialResult.getId() > currentMaxId) {
                currentMaxId = Long.valueOf(tDialResult.getId());
            }
        }

        // 存入Redis
        redisTemplate.opsForValue().set("lastMaxId", currentMaxId.toString());
        return String.join(",", keysList);
    }

    public CompletableFuture<Integer> sendMaulMessageAsync(int count) {
        // 增加线程数以加大服务器压力
        int threadCount = 10; // 可根据需要调整线程数


        TCrmContChildState tCrmContChildState = jpaEntityService.findEntityById(581);
        AtomicInteger id = new AtomicInteger(1);
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger failureCount = new AtomicInteger(0);
        String value = RocketmqConstant.MyTopic().get(RocketmqConstant.TopicEnum.MANUAL_DIAL_RESULT_MSG_TOPIC);

        // 创建线程池以实现更高并发
        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);

        List<CompletableFuture<Void>> futures = new ArrayList<>();

        // 将任务分配给多个线程
        int messagesPerThread = count / threadCount;
        for (int t = 0; t < threadCount; t++) {
            final int threadId = t;
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                int start = threadId * messagesPerThread;
                int end = (threadId == threadCount - 1) ? count : (threadId + 1) * messagesPerThread;

                for (int i = start; i < end; i++) {
                    try {
//                        int currentId = id.incrementAndGet();
                        tCrmContChildState.setId(i);
                        rocketMQTemplate.send(value, MessageBuilder.withPayload(JSON.toJSONString(tCrmContChildState))
                                .setHeader("KEYS", String.format("%s%s", new SimpleDateFormat("yyyyMMddHHmmss").format(new java.util.Date()), RandomStringUtils.randomAlphabetic(18).toUpperCase())).build());
                        successCount.incrementAndGet();
                    } catch (Exception e) {
                        failureCount.incrementAndGet();
                        logger.error("发送消息失败", e);
                    }
                }
            }, executorService);

            futures.add(future);
        }

        // 等待所有线程完成
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

        return allFutures.thenApply(v -> {
            executorService.shutdown(); // 关闭线程池
            return successCount.get();
        }).exceptionally(e -> {
            logger.error("执行过程中发生错误", e);
            executorService.shutdown();
            return failureCount.get();
        });
    }


    public CompletableFuture<Integer> sendHNMessageAsync(Integer count) {
        final Semaphore mqSemaphore = new Semaphore(30);
        HnCrmContChildState hnCrmContChildState = jpaEntityService.findHnEntityById(3357359);
        AtomicInteger id = new AtomicInteger(8000000);
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger failureCount = new AtomicInteger(0);
        String value = RocketmqConstant.MyTopic().get(RocketmqConstant.TopicEnum.HNBOC_SEAT_TASK_ASSIGN_LOG_TOPIC);

        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            for (int i = count; i > 0; i--) {
                try {
                    mqSemaphore.acquire();
                    int currentId = id.incrementAndGet();
                    hnCrmContChildState.setId(currentId);
                    rocketMQTemplate.send(value, MessageBuilder.withPayload(JSON.toJSONString(hnCrmContChildState))
                            .setHeader("KEYS", String.format("%s%s", new SimpleDateFormat("yyyyMMddHHmmss").format(new java.util.Date()), RandomStringUtils.randomAlphabetic(18).toUpperCase())).build());
                    successCount.incrementAndGet();
                } catch (Exception e) {
                    failureCount.incrementAndGet();
                    logger.error("发送消息失败", e);
                } finally {
                    mqSemaphore.release();
                }
            }
        });

        return future.thenApply(v -> {
            try {
                future.get();
                return successCount.get();
            } catch (InterruptedException | ExecutionException e) {
                logger.error("获取异步结果失败", e);
                return failureCount.get();
            }
        });
    }

    public CompletableFuture<Integer> sendGjBbMessageAsync(Integer count) {
        String value = RocketmqConstant.MyTopic().get(RocketmqConstant.TopicEnum.ROBOT_DIAL_RESULT_MSG_TOPIC);
        String dialTaskMQMsg = "{\"accountId\":501,\"data\":[{\"bankOrg\":\"47000\",\"bankOrgName\":\"兴业银行昆明分行\",\"batchDetailId\":\"16882100\",\"batchId\":\"6905\",\"batchName\":\"拨测批次20250407143458\",\"blacklist\":false,\"callRecordPath\":\"/sound/20250407/06/057100204017339762006-7441b0304950495897aa4bbd99d97f97-1744007724726.oga\",\"callType\":2,\"caller\":\"057100204\",\"dataSourceType\":1,\"dialDuration\":56,\"dialId\":829208,\"dialStatus\":1,\"dialStatusDesc\":\"已接通\",\"dialTime\":\"2025-04-07 14:35:22\",\"execLogs\":[{\"content\":\"喂（空1.2秒）您好，这里是珠海建行，您在我行有笔定期存款已经到期还未办理转存，我安排网点客户经理给您介绍下最新转存产品信息好吗？!(R130)\",\"flowNode\":{\"needHangup\":false,\"nodeId\":\"df831ec0-48dc-11ec-8ea2-2766b20a6b2d\"},\"robot\":true,\"time\":\"2025-04-07 14:35:37\"},{\"content\":\"静音(S)\",\"robot\":false,\"time\":\"2025-04-07 14:35:59\"},{\"content\":\"您好，能听到吗？(R2)\",\"robot\":true,\"time\":\"2025-04-07 14:35:59\"},{\"content\":\"静音(S)\",\"robot\":false,\"time\":\"2025-04-07 14:36:12\"},{\"content\":\"您好，请问能听到我的声音吗？(R3)\",\"robot\":true,\"time\":\"2025-04-07 14:36:12\"},{\"content\":\"静音(S)\",\"robot\":false,\"time\":\"2025-04-07 14:36:25\"},{\"content\":\"抱歉，暂时无法听到您的声音，这边就不打扰您了，再见！(R4)\",\"robot\":true,\"time\":\"2025-04-07 14:36:25\"},{\"content\":\"主叫挂断(RE)\",\"robot\":true,\"time\":\"2025-04-07 14:36:32\"}],\"extraInfo\":{\"客户所属银行机构号\":\"47000\",\"联系方式\":\"17339762006\"},\"followUserId\":63488,\"hangupTime\":\"2025-04-07 14:36:33\",\"intentionClassify\":\"ac0fae9e-ad31-423f-9487-0012c9b6b9da\",\"intentionClassifyDesc\":\"漠不关心\",\"phone\":\"17339762006\",\"pickupTime\":\"2025-04-07 14:35:37\",\"ringDuration\":15,\"robotId\":258,\"roundCount\":0,\"sceneId\":169,\"sceneLabels\":[\"生命\",\"紧急\",\"阿萨给\"],\"sceneName\":\"报表1\",\"sceneNum\":\"20250326XNCAKGQS\",\"seatConversingDuration\":0,\"tag\":\"全程静音,开场白听完,[开场白1]听完静音,连续静音,转接失败\",\"taskCompleteStatus\":1,\"taskDetailId\":\"20250407143458501-qkyeadoi\",\"taskId\":\"20250407143458501\",\"taskName\":\"拨测批次20250407143458\",\"transferResult\":4}],\"logId\":\"9f843118-c077-4ddc-8705-ad91b710e406\"}";
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger failureCount = new AtomicInteger(0);
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            for (int i = count; i > 0; i--) {
                try {
                    rocketMQTemplate.send(value, MessageBuilder.withPayload(JSON.toJSONString(dialTaskMQMsg))
                            .setHeader("KEYS", String.format("%s%s", new SimpleDateFormat("yyyyMMddHHmmss").format(new java.util.Date()), RandomStringUtils.randomAlphabetic(18).toUpperCase())).build());
                    successCount.incrementAndGet();
                } catch (Exception e) {
                    failureCount.incrementAndGet();
                    logger.error("发送消息失败", e);
                }
            }
        });

        return future.thenApply(v -> {
            try {
                future.get();
                return successCount.get();
            } catch (InterruptedException | ExecutionException e) {
                logger.error("获取异步结果失败", e);
                return failureCount.get();
            }
        });
    }
}